package com.itcast.spark.baseTree

import java.io.File

import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}


/**
 * DESC:思路分析
 * 1-准备环境
 * 2-读取数据
 * 3-准备特征工程(这里libsvm是spark官方提供的数据集是已经做完特征工程处理的数据集)
 * 4-准备算法
 * 5-训练模型
 * 6-模型保存
 */
object SparkLibSvmDtcModel {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("SparkLibSvmDtcModel").setMaster("local[*]")
    val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //2-读取数据
    val dataDF: DataFrame = spark.read.format("libsvm").load("./datasets/mldata/sample_libsvm_data.txt")
    //自带features+label
    dataDF.show()
    dataDF.printSchema()
    //3-准备特征工程
    val split: Array[Dataset[Row]] = dataDF.randomSplit(Array(0.8, 0.2), seed = 123L)
    val trainingData: Dataset[Row] = split(0)
    val testData: Dataset[Row] = split(1)
    //4-准备算法=-======超参数的部分
    val classifier: DecisionTreeClassifier = new DecisionTreeClassifier()
      .setFeaturesCol("features")
      .setLabelCol("label")
      .setMaxDepth(5) //一般在5层左右即可
      .setMinInstancesPerNode(1)
      .setImpurity("entropy")
      .setMaxBins(32) //连续值离散化的最大的分箱的个数
      .setMinInfoGain(0.0)
      .setPredictionCol("PredictionCol") //用户自己定义的预测列
      .setProbabilityCol("ProbabilityCol")
    //5-训练模型
    val model: DecisionTreeClassificationModel = classifier.fit(trainingData)
    val y_train_pred: DataFrame = model.transform(trainingData)
    val y_test_pred: DataFrame = model.transform(testData)
    //6-引入-判断分类效果的量-----
    val evaluator: MulticlassClassificationEvaluator = new MulticlassClassificationEvaluator()
      .setPredictionCol("PredictionCol")
      .setMetricName("accuracy")
      .setLabelCol("label")
    val accuracy_train: Double = evaluator.evaluate(y_train_pred)
    val accuracy_test: Double = evaluator.evaluate(y_test_pred)
    println("train accuracy value is:", accuracy_train)
    println("test accuracy value is:", accuracy_test)
    //    (train accuracy value is:,1.0)
    //    (test accuracy value is:,0.9090909090909091)
    //6-模型保存
    val path = "./datasets/mldata/modelSave1"
    val file = new File(path)
    //如果文件存在
    if (file.exists()) {
      println("存在!直接预测")
      DecisionTreeClassificationModel.load("./datasets/mldata/modelSave1").transform(testData).show()
    } else {
      println("不存在!")
      model.save(path)
    }
  }
}